package org.yuanzheng.state

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.yuanzheng.source.StationLog

/**
 * 统计每个手机的呼叫时间间隔
 */
object KeyedState2 {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val path = getClass.getResource("/station.log").getPath
    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.readTextFile(path).map(line => {
      val split = line.split(",")
      StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
    })
    //状态有两种情况
    stream.keyBy(_.callOut).mapWithState[(String, Long), StationLog] {
      case (in: StationLog, None) => ((in.callOut, 0), Some(in)) //状态中没有值
      case (in: StationLog, pre: Some[StationLog]) =>
        val interval = in.callTime - pre.get.callTime
        ((in.callOut, interval), Some(in))
    }.filter(_._2 != 0).print()
    streamEnv.execute()
  }
}

